package com.atguigu.flink.chapter08_exec2;

import com.atguigu.flink.pojo.MyUtil;
import com.atguigu.flink.pojo.UserBehavior;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2022/10/29
 *
 *      读数据的时候，并行度设置为1.
 *
 *          基于事件时间，从bean中抽取时间属性(毫秒)
 *
 *      -------------------
 *          每小时统计，过去1h的pv。
 *              如果过滤出了所有的pv，无需keyBy
 *              基于事件时间的滚动窗口，窗口的大小为1h。
 *                      基于个数:  countWindowAll
 *                      基于时间:   windowAll
 *
 *              计算方式就是累加求和：
 *                      增量: sum,max,min... reduce,aggregate
 *                              reduce： 输入和输出类型必须是一致的
 *                              aggregate: 输入和输出类型不一致的
 *                      全量: process
 *
 *                当前输入: UserBehavior
 *                输出:    时间+ PV
 *
 窗口[2017-11-26 09:00:00,2017-11-26 10:00:00):41890
 窗口[2017-11-26 10:00:00,2017-11-26 11:00:00):48022
 窗口[2017-11-26 11:00:00,2017-11-26 12:00:00):47298
 窗口[2017-11-26 12:00:00,2017-11-26 13:00:00):44499
 窗口[2017-11-26 13:00:00,2017-11-26 14:00:00):48649
 窗口[2017-11-26 14:00:00,2017-11-26 15:00:00):50838
 窗口[2017-11-26 15:00:00,2017-11-26 16:00:00):52296
 窗口[2017-11-26 16:00:00,2017-11-26 17:00:00):52552
 窗口[2017-11-26 17:00:00,2017-11-26 18:00:00):48292
 窗口[2017-11-26 18:00:00,2017-11-26 19:00:00):13
 */
public class Demo1_PV
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<UserBehavior> watermarkStrategy = WatermarkStrategy.<UserBehavior>forMonotonousTimestamps()
            .withTimestampAssigner((u, ts) -> u.getTimestamp());

        env.readTextFile("data/UserBehavior.csv").setParallelism(1)
           .map(new MapFunction<String, UserBehavior>()
           {
               @Override
               public UserBehavior map(String value) throws Exception {
                   String[] words = value.split(",");
                   return new UserBehavior(
                       Long.valueOf(words[0]),
                       Long.valueOf(words[1]),
                       Integer.valueOf(words[2]),
                       words[3],
                       Long.valueOf(words[4]) * 1000
                   );
               }
           })
           //过滤出pv
           .filter(u -> "pv".equals(u.getBehavior()))
           .assignTimestampsAndWatermarks(watermarkStrategy)
           .windowAll(TumblingEventTimeWindows.of(Time.hours(1)))
           .aggregate(new AggregateFunction<UserBehavior, Integer, Integer>()
           {
               @Override
               public Integer createAccumulator() {
                   return 0;
               }

               @Override
               public Integer add(UserBehavior value, Integer acc) {
                   return acc += 1;
               }

               @Override
               public Integer getResult(Integer acc) {
                   return acc;
               }

               @Override
               public Integer merge(Integer a, Integer b) {
                   return null;
               }
           }, new ProcessAllWindowFunction<Integer, String, TimeWindow>()
           {
               @Override
               public void process(Context context, Iterable<Integer> elements, Collector<String> out) throws Exception {
                   TimeWindow window = context.window();
                   out.collect(MyUtil.printTimeWindow(window) + ":" + elements.iterator().next());
               }
           }).print().setParallelism(1);


        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }

    }
}
